use std::{
	collections::HashMap,
	sync::{
		atomic::{AtomicU64, Ordering},
		LazyLock,
	},
	time::Duration,
};

use bytes::{Bytes, BytesMut};
use futures::future::BoxFuture;
use tokio::{
	sync::{mpsc, RwLock},
	time,
};
use tools::utils::gen_id::get_now_duration;
use tracing::{info, trace};

use crate::sip2::{
	config::sip_config,
	dialog::{sipdata::SipData, LENGTH_ZERO},
	head::{CONTENT_LENGTH, WWW_AUTHENTICATE},
	register::{load_auth_field, verify},
	transport::TransportProtocol,
	utils::{generate_disorder_u8s, get_bytes_str, WriteSip},
	SipResponse,
};

use super::{SipRegEvent, SipRegisterUser};

pub static USER_HOLD: LazyLock<RwLock<HashMap<String, SipRegisterUser>>> =
	LazyLock::new(|| RwLock::new(HashMap::new()));

///通过已经注册的用户名,获取用户传输对象
pub(crate) async fn load_user_transport(user_name: &str) -> Option<TransportProtocol> {
	let users = USER_HOLD.read().await;

	users.get(user_name).map(|it| it.tp.clone())
}

pub(crate) async fn register_manage<F, F2>(
	mut recv_command: mpsc::Receiver<(SipData, TransportProtocol)>,
	handler_event: &F,
	get_user_passwd: F2,
) where
	F: Fn(SipRegEvent) -> () + Send + Sync,
	F2: Fn(&str) -> BoxFuture<Option<String>> + 'static + Send + Sync,
{
	let sleep = time::sleep(Duration::from_secs(20));
	tokio::pin!(sleep);
	//只使用一个超时器.超时原因,在打开的时候进行判断
	loop {
		sleep.set(time::sleep(Duration::from_secs(20)));
		tokio::select! {
			biased;
			Some((sip_data, tp)) = recv_command.recv() => {
				let auth_bytes = sip_data.authorization.bytes();
				//返回401
				if auth_bytes.is_empty() {
					tp.send(generate_401(&sip_data));
					continue;
				}

				let auth_info = load_auth_field(auth_bytes);
				let users = USER_HOLD.read().await;
				//先通过已经存在的用户获取完整的用户信息.如果不存在,则通过调用业务逻辑获取
				let user = match users.get(auth_info.username){
					Some(u) => {
						let mut u = u.clone();
						u.tp = tp;
						u.expires_timestamp = get_now_duration() + Duration::from_secs(get_bytes_str(sip_data.expires.bytes()).parse().unwrap_or(0));

						u
					}
					None => match get_user_passwd(&auth_info.username.to_owned()).await {
						Some(pw) => SipRegisterUser::new(
							pw,
							Duration::from_secs(get_bytes_str(sip_data.expires.bytes()).parse().unwrap_or(0)),
							auth_info.realm.to_owned(),
							tp,
							),
						None => {
							handler_event(SipRegEvent::RegistFailure(auth_info.username.to_owned()));
							tp.send(generate_403(&sip_data));
							continue;
						}
					}
				};
				drop(users);

				if verify(&auth_info, &user) {
					user.tp.send(generate_200(&sip_data));
					let mut users_w = USER_HOLD.write().await;
					//是否是登出..
					if has_login_out(&user) {
						trace!("用户登出:{:?}",user);
						handler_event(SipRegEvent::RegistLoginOut(user));
						users_w.remove(auth_info.username);
					} else {
						trace!("用户登录:{:?}",user);
						handler_event(SipRegEvent::RegistSuccess(user.clone()));
						users_w.insert(auth_info.username.to_owned(), user);
					}
				} else {
					user.tp.send(generate_403(&sip_data));
				}
			}
			//处理超时状态
			_ = &mut sleep => {
				// 循环users.判断是否超时移出
				let mut us = USER_HOLD.write().await;
				us.retain(|_,v| !v.overtime());

				trace!("清理完成当前注册用户...剩余数量:{}",us.len());
			}
			else => {
				break;
			}
		}
	}

	info!("结束运行!");
}

///判断是否登出,,sip的expires字段,如果小于当前时间,则表示登出
fn has_login_out(user: &SipRegisterUser) -> bool {
	user.expires_timestamp < get_now_duration()
}

fn generate_401(sip_data: &SipData) -> Bytes {
	static WWW_AUTH: AtomicU64 = AtomicU64::new(120531444340919636u64);
	let mut buf = BytesMut::with_capacity(1024);

	buf.put_sip_response(&SipResponse::Unauthorized);
	buf.put_unbroken_head(&sip_data.via);
	buf.put_unbroken_head(&sip_data.from);
	buf.put_unbroken_head(&sip_data.to);
	buf.put_unbroken_head(&sip_data.call_id);
	buf.put_unbroken_head(&sip_data.cseq);

	buf.put_head_fn(WWW_AUTHENTICATE, |buf| {
		buf.extend_from_slice(b"Digest realm=\"");
		buf.extend_from_slice(sip_config().domain.as_bytes());
		buf.extend_from_slice(b"\", nonce=\"");
		buf.extend_from_slice(&generate_disorder_u8s(
			WWW_AUTH.fetch_add(12907525553771931, Ordering::Relaxed),
		));
		buf.extend_from_slice(&generate_disorder_u8s(
			WWW_AUTH.fetch_add(12907525553771931, Ordering::Relaxed),
		));
		buf.extend_from_slice(b"\", algorithm=MD5, qop=\"auth\"");
	});
	buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
	buf.put_heads_end();

	buf.freeze()
}

fn generate_200(sip_data: &SipData) -> Bytes {
	let mut buf = BytesMut::with_capacity(1024);

	buf.put_sip_response(&SipResponse::Ok);
	buf.put_unbroken_head(&sip_data.via);
	buf.put_unbroken_head(&sip_data.from);
	buf.put_unbroken_head(&sip_data.to);
	buf.put_unbroken_head(&sip_data.call_id);
	buf.put_unbroken_head(&sip_data.cseq);
	buf.put_unbroken_head(&sip_data.contact);
	buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
	buf.put_heads_end();

	buf.freeze()
}
fn generate_403(sip_data: &SipData) -> Bytes {
	let mut buf = BytesMut::with_capacity(1024);

	buf.put_sip_response(&SipResponse::Forbidden);
	buf.put_unbroken_head(&sip_data.via);
	buf.put_unbroken_head(&sip_data.from);
	buf.put_unbroken_head(&sip_data.to);
	buf.put_unbroken_head(&sip_data.call_id);
	buf.put_unbroken_head(&sip_data.cseq);
	buf.put_head_field(CONTENT_LENGTH, LENGTH_ZERO);
	buf.put_heads_end();

	buf.freeze()
}
